package com.lwl.java.base;

import com.lwl.Utils.SleepUtil;

import java.util.LinkedList;
import java.util.Random;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @author lwl
 * @Description TODO
 * @date 2025/9/7 11:43
 */
public class ConsumerAndProducer {
    class Message {
        String info;

        public Message(String info) {
            this.info = info;
        }

        @Override
        public String toString() {
            return "Message{" +
                    "info='" + info + '\'' +
                    '}';
        }
    }

    private LinkedList<Message> queue = new LinkedList<>();

    private Lock lock = new ReentrantLock();

    private Condition notFull = lock.newCondition();

    private Condition notEmpty = lock.newCondition();

    private final int MAX_CAPACITY = 10;

    private volatile boolean isRunning = false;

    public Message getMessageSync() {
        lock.lock();
        try {
            while (queue.isEmpty()) {
                notEmpty.await();
            }
            Message msg = queue.pollFirst();
            System.out.println("当前消息数为 ：" + queue.size());
            notFull.signalAll();
            return msg;
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        } finally {
            lock.unlock();
        }
    }

    public void addMessageSync(Message message) {
        lock.lock();
        try {
            while (queue.size() >= MAX_CAPACITY) {
                boolean addSuccess = notFull.await(5, TimeUnit.SECONDS);
                if (!addSuccess) {
                    System.out.println("消息" + message + "投递失败 ～");
                    return;
                }
            }
            queue.addLast(message);
            System.out.println("当前消息数为 ：" + queue.size());
            notEmpty.signalAll();
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        } finally {
            lock.unlock();
        }
    }

    public void startConsume() {
        Thread consumerThread = new Thread(() -> {
            while (isRunning) {
                Message msg = getMessageSync();
                if (msg != null) {
                    System.out.println(Thread.currentThread().getName() + " 开始消费消息 ： " + msg);
                    SleepUtil.sleep(1000);
                    System.out.println(Thread.currentThread().getName() + " 结束消费消息 ： " + msg);
                }
            }
            System.out.println(Thread.currentThread().getName() + " 结束消费 ～ ");
        });
        consumerThread.start();
    }

    public void startProduce() {
        Thread producerThread = new Thread(() -> {
            while (isRunning) {
                Message msg = new Message(String.valueOf(new Random().nextInt()));
                System.out.println(Thread.currentThread().getName() + " 开始生产消息 ： " + msg);
                SleepUtil.sleep(200);
                addMessageSync(msg);
                System.out.println(Thread.currentThread().getName() + " 投递消息结束 ： " + msg);
            }
            System.out.println(Thread.currentThread().getName() + " 结束生产 ～ ");
        });
        producerThread.start();
    }

    public static void main(String[] args) {
        ConsumerAndProducer cp = new ConsumerAndProducer();
        cp.isRunning = true;
        cp.startConsume();
        cp.startConsume();
        cp.startProduce();
        SleepUtil.sleep(10000);
        cp.isRunning = false;
    }
}
